package com.exsun.gateway.jt809.slave.handler;

import com.exsun.gateway.jt809.common.LinkType;
import com.exsun.gateway.jt809.kafka.KafkaProducer;
import com.exsun.gateway.jt809.data.PackageData;
import com.exsun.gateway.jt809.common.LinkMgr;
import com.exsun.gateway.jt809.handler.ResponseHandler;
import exsun.jt808.data.jt809_data.JT809LinkStatus;
import io.netty.buffer.ByteBufUtil;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
@Sharable
public class JT809ServerHandler extends SimpleChannelInboundHandler<PackageData> { // (1)
	private final Logger logger = Logger.getLogger(JT809ServerHandler.class);
	@Autowired
	ResponseHandler responseHandler;

	@Autowired
	KafkaProducer kafkaProducer;

	@Override
	protected void channelRead0(ChannelHandlerContext ctx, PackageData pkg) throws Exception {
		try {
            logger.info("收到来自客户端的请求数据：" + ByteBufUtil.hexDump(pkg.getMsgBodyBytes()));
			PackageData pak = responseHandler.response(pkg);
			if(pak == null)return;
			LinkMgr.push2FitChannel(pak.getMsgHeader().getMsgId(), pak.getMsgBodyBytes(), pak.getLinkType());
		}finally {
			release(pkg);
		}
	}
	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { // (4)
		logger.info("从链路通讯通道发生异常");
		cause.printStackTrace();
	}

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		logger.info("从链路通道连接 remoteAddress： + "  + ctx.channel().remoteAddress());

		LinkMgr.setSlaveChannel(ctx.channel());
		LinkMgr.setSlaveChannelLoggingStatus(true);
		//将链路状态推送到kafka
		kafkaStatus(JT809LinkStatus.SlaveOnline);

	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		logger.info("从链路通道断开连接");
		LinkMgr.setSlaveChannel(null);
		LinkMgr.setSlaveChannelLoggingStatus(false);
		//将链路状态推送到kafka
		kafkaStatus(JT809LinkStatus.SlaveOffline);
	}

	@Override
	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
		if (IdleStateEvent.class.isAssignableFrom(evt.getClass())) {
			IdleStateEvent event = (IdleStateEvent) evt;
			if ((event.state() == IdleState.READER_IDLE)) {
				logger.error("从链路读数据超时");
				ctx.close();
			}
		}
	}

	private void release(Object msg) {
		try {
			ReferenceCountUtil.release(msg);
		} catch (Exception e) {
			logger.error(e.getMessage(),e);
			e.printStackTrace();
		}
	}

	private void kafkaStatus(JT809LinkStatus status){
		kafkaProducer.kafkaStatus(status);
	}

}